package com.jida.hadoop.mr.datamatch;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.TreeMap;
 
import com.jida.hadoop.mr.tools.*;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
 
public class MapJoinMapper extends Mapper<LongWritable, Text, NullWritable, Text> {
    private Text ovalue = new Text();
    //Map<一级域名, TreeMap<url的拼接字符串, 空>>，用treemap按优先级排序
    private Map<String, TreeMap<String, String>> joinData = new HashMap<String, TreeMap<String, String>>();
 
    /**
     * setup在map之前执行，加载地址库
     * */
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        // 预处理把要关联的文件加载到缓存中
        // 我们这里只缓存了一个文件，所以取第一个即可，创建BufferReader去读取
        Path file = new Path(context.getCacheFiles()[0].getPath());
        BufferedReader reader = new BufferedReader(new FileReader(file.toString()));
 
        String str = "";
        try {
            // 一行一行读取
            while ((str = reader.readLine()) != null) {
                // 对缓存中的表进行分割
                String[] splits = str.split(TProperties.getValue("fileoutsplit"));
                //map是否包含一级域名
                if (joinData.containsKey(splits[1])) {
                    //增加treemap中url值：匹配地址,匹配级别,行为ID,产品标识,预购类型
                    joinData.get(splits[1]).put(splits[2] + "," + splits[splits.length-3] + "," + splits[0] +","+ splits[splits.length-2]+ "," + splits[splits.length-1], "");
                } else {
                    //创建map中key值，添加treemap，StringComparator修改treemap排序方式
                    TreeMap<String, String> treemap = new TreeMap<String, String>( new StringComparator());
                    //url地址,匹配级别,行为ID,是否产品,预购类型
                    treemap.put(splits[2] + "," + splits[splits.length-3]+ "," + splits[0] +","+ splits[splits.length-2] + "," + splits[splits.length-1], "");
                    //key一级域名
                    joinData.put(splits[1], treemap);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            reader.close();
        }
    }
 
    /**
     * map端循环匹配数据，不使用reduce部分提高处理效率。
     * */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 获取从HDFS中加载的表
        String[] values = value.toString().split(TProperties.getValue("fileoutsplit"));
        //youku.com
        if (joinData.containsKey(values[1])) {
            //获取一级域名下匹配地址(joinData集合中key对应的value值)
            TreeMap<String, String> treeMap1 = joinData.get(values[1]);
            @SuppressWarnings("rawtypes")
			Iterator iterator = treeMap1.keySet().iterator();//获取treeMap1集合中所有的key对象，并转换为迭代器
            //按级别循环匹配
            while (iterator.hasNext()) {
                String[] krule = ((String) iterator.next()).split(",");
                //values[2] = http://push.m.youku.com/feedback/recv/BT_713852830_436662@436662?pid
                //values[2]和地址库treeMap的所有url(二级或三级域名)每个匹配
                if (values[2].indexOf(krule[0]) >= 0) {
                    StringBuffer sb = new StringBuffer();
                    //行为id，用户号码，是否产品，url,预购类型
                    sb.append(krule[2]).append(TProperties.getValue("outfilesplit"))
                            .append(values[0]).append(TProperties.getValue("outfilesplit"))
                            .append(krule[3]).append(TProperties.getValue("outfilesplit"))
                            .append(values[2]).append(TProperties.getValue("outfilesplit"))
                            .append(krule[4]);
                    ovalue.set(sb.toString());
                    //匹配后输出并退出匹配循环
                    context.write(NullWritable.get(), ovalue);
                    return ;
                }
            }
        }
    }
}
